package logic

import (
	"fmt"
	"sync"
	"time"

	"gddgame.cc/galaxy/asteroid"
	"gddgame.cc/galaxy/asteroid/strategy/hash"
)

type ObserverHandler func(observer string)
type ObserverNodeHandler func(observer string, node string, online bool)

// Watcher等待Observer的通知，实现服务发现
type Watcher interface {
	OnObserver(handler ObserverHandler)
	OnObserverNode(handler ObserverNodeHandler)
	NodeList(observer string) []string
	AllocNode(observer string, key string) string
	NodeAll() []string
	Observing(observer string) *asteroid.Publisher
	Observers(t string) map[string][]string
	Meta(node string) map[string]interface{}
	Watching(event interface{}, fn asteroid.OnHandler)
	AllocObserver(observer string) string
}
type nodeInfo struct {
	observer string
	meta     map[string]interface{}
}
type WatcherInstance struct {
	asteroid.Channel

	observers sync.Map
	nodes     sync.Map

	observerFunc     []ObserverHandler
	observerNodeFunc []ObserverNodeHandler

	aliveFunc []ObserverNodeHandler

	subscriber *asteroid.Subscriber
	publisher  *asteroid.Publisher
	strategy   asteroid.Strategy

	id string
}

func NewWatcher(id string, core string, channel asteroid.Channel) *WatcherInstance {
	watcher := &WatcherInstance{
		Channel:   channel,
		observers: sync.Map{},
		nodes:     sync.Map{},

		observerFunc:     []ObserverHandler{},
		observerNodeFunc: []ObserverNodeHandler{},
		aliveFunc:        []ObserverNodeHandler{},
		id:               id,
	}

	watcher.subscriber = watcher.Subscribe(id)
	publisher, ok := watcher.Publish(core)
	if !ok {
		publisher.SetStrategy(hash.NewStrategy())
	}
	watcher.publisher = publisher

	// 按最小频次调度
	go func() {
		now := time.Now()
		// 心跳：
		//    1. 确保快速重启的observer收到observer通知：因为adapt未能触发
		//    2. 为report提供alive接口，监控alive
		<-time.After(time.Second * time.Duration(60-now.Second()))
		ticker := time.NewTicker(time.Second * 30)
		for _ = range ticker.C {
			watcher.sendAlive()
		}
	}()
	watcher.Watching("watcher", func(message asteroid.Message) {
		observer := message.GetOrigin()
		//fmt.Println("watcher", observer)
		meta := message.Data()
		info := &nodeInfo{
			observer: observer,
			meta:     meta.(map[string]interface{}),
		}
		watcher.nodes.Store(message.GetFrom(), info)
		if _, ok := watcher.observers.Load(observer); !ok {
			//fmt.Println("init observer", observer)
			publisher, ok := watcher.Publish(observer)
			if !ok {
				publisher.SetStrategy(hash.NewStrategy())
			}
			watcher.observers.Store(observer, publisher)

			//fmt.Println("watcher store", observer)
			for key, _ := range watcher.observerFunc {
				watcher.observerFunc[key](observer)
			}

			publisher.OnJoin(func(node string) {
				for key, _ := range watcher.observerNodeFunc {
					watcher.observerNodeFunc[key](observer, node, true)
				}
			})

			publisher.OnLeave(func(node string) {
				for key, _ := range watcher.observerNodeFunc {
					watcher.observerNodeFunc[key](observer, node, false)
				}
			})
		}
		message.Response(true)
	})

	publisher.OnJoin(func(node string) {
		// 请求watch
		fmt.Println("=>observer meta", node)
		publisher.Direct(node).OriginId(id).Send("observer")
	})
	publisher.OnLeave(func(node string) {
		// 移除节点信息
		watcher.nodes.Delete(node)
	})
	return watcher
}

func (watcher *WatcherInstance) sendAlive() {
	watcher.publisher.All().OriginId(watcher.id).Callback(func(success bool, data map[string]interface{}) {
		for node := range data {
			online := false
			if data[node] != nil {
				online = true
			}
			value, ok := watcher.nodes.Load(node)
			if ok {
				for key, _ := range watcher.aliveFunc {
					watcher.aliveFunc[key](value.(*nodeInfo).observer, node, online)
				}
			}
		}
	}).Send("observer")
}

// 需要避免和agent的subscriber绑定相同的事件
// 对于包含core的服务，watcher.Watching == observer.subscriber
func (watcher *WatcherInstance) Watching(event interface{}, fn asteroid.OnHandler) {
	watcher.subscriber.On(event, fn)
}
func (watcher *WatcherInstance) OnObserver(handler ObserverHandler) {
	watcher.observerFunc = append(watcher.observerFunc, handler)
}
func (watcher *WatcherInstance) OnObserverNode(handler ObserverNodeHandler) {
	watcher.observerNodeFunc = append(watcher.observerNodeFunc, handler)
}
func (watcher *WatcherInstance) OnAlive(handler ObserverNodeHandler) {
	watcher.aliveFunc = append(watcher.aliveFunc, handler)
}

func (watcher *WatcherInstance) NodeList(observer string) []string {
	if value, ok := watcher.observers.Load(observer); ok {
		publisher := value.(*asteroid.Publisher)
		//fmt.Println("node observer", observer, publisher.Nodes())
		return publisher.OtherNodes()
	} else {
		//fmt.Println("node observer", observer, " miss")
		return []string{}
	}
}

func (watcher *WatcherInstance) AllocNode(observer string, key string) string {
	if value, ok := watcher.observers.Load(observer); ok {
		publisher := value.(*asteroid.Publisher)
		//fmt.Println("node observer", observer, publisher.Nodes())
		return publisher.Alloc(key)
	} else {
		//fmt.Println("node observer", observer, " miss")
		return ""
	}
}

func (watcher *WatcherInstance) NodeAll() []string {
	list := []string{}
	watcher.nodes.Range(func(key, value interface{}) bool {
		list = append(list, key.(string))
		return true
	})
	return list
}

func (watcher *WatcherInstance) Observing(observer string) *asteroid.Publisher {
	if value, ok := watcher.observers.Load(observer); ok {
		publisher := value.(*asteroid.Publisher)
		return publisher
	} else {
		return nil
	}
}

func (watcher *WatcherInstance) Observers(t string) map[string][]string {
	result := make(map[string][]string)
	watcher.observers.Range(func(key, value interface{}) bool {
		publisher := value.(*asteroid.Publisher)
		nodes := publisher.OtherNodes()
		if len(nodes) == 0 {
			return true
		}
		if t != "" {
			meta := watcher.Meta(nodes[0])
			if meta == nil {
				return true
			}
			//fmt.Println(meta)
			if meta["type"].(string) != t {
				return true
			}
		}
		result[key.(string)] = nodes
		return true
	})
	return result
}
func (watcher *WatcherInstance) Meta(node string) map[string]interface{} {
	value, ok := watcher.nodes.Load(node)
	//fmt.Println("load", node, value)
	if ok {
		return value.(*nodeInfo).meta
	}
	return nil
}

// 以System的节点作为分配，Core是所有机器节点
func (watcher *WatcherInstance) SetStrategy(strategy asteroid.Strategy) {
	watcher.strategy = strategy
}

func (watcher *WatcherInstance) AllocObserver(observer string) string {
	return watcher.strategy.Alloc(observer)
}
